package com.heima.kafka.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        stream.flatMapValues(value -> {
            //空格解析
            String[] s = value.split(" ");
            return Arrays.asList(s);
        }).groupBy((key,value) -> value)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .count().toStream().map((key,value) -> new KeyValue<>(key.key(),String.valueOf(value)))
                .to("itcast-topic-output");

        return stream;
    }
}